package bx.actor;

import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
import akka.http.javadsl.Http;
import akka.http.javadsl.marshallers.jackson.Jackson;
import akka.http.javadsl.model.*;
import bx.InteractionEntity.CommonResponse;
import bx.InteractionEntity.WorkerRegistRequest;
import bx.JobWorkerBootstrap;
import com.google.gson.Gson;
import lombok.Builder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;


public class WorkerRegistActor extends AbstractBehavior<WorkerRegistActor.WorkerRegistActorMsg> {
    private final static Logger log = LoggerFactory.getLogger(WorkerRegistActor.class);
    public static final String REGIST_PATH = "/regist";
    private static final WorkerRegistStartMsg WORKER_REGIST_START_MSG = WorkerRegistStartMsg.builder().build();

    // 在注册成功后，成为该 actor 的子 actor，完成其心跳操作，仅注册成功后有效
    private ActorRef<WorkerHeartbeatActor.WorkerHeartbeatActorMsg> workerHeartbeatActorActor;
    private final List<String> bxJobServerList;
    private int currentIndex;


    public interface WorkerRegistActorMsg {
    }

    @Builder
    public static class WorkerRegistStartMsg implements WorkerRegistActorMsg {
    }

    @Builder
    public static class WorkerRegistSuccessMsg implements WorkerRegistActorMsg {
    }

    public WorkerRegistActor(ActorContext<WorkerRegistActor.WorkerRegistActorMsg> context, List<String> bxJobServerList) {
        super(context);
        this.bxJobServerList = bxJobServerList;
        context.getSelf().tell(WORKER_REGIST_START_MSG);
    }

    public static Behavior<WorkerRegistActor.WorkerRegistActorMsg> create(List<String> bxJobServerList) {
        return Behaviors.setup(context -> new WorkerRegistActor(context, bxJobServerList));
    }

    @Override
    public Receive<WorkerRegistActor.WorkerRegistActorMsg> createReceive() {
        return newReceiveBuilder().onMessage(WorkerRegistActor.WorkerRegistStartMsg.class, this::onStartRegistMsg).onMessage(WorkerRegistActor.WorkerRegistSuccessMsg.class, WorkerRegistStartMsg -> {
            workerHeartbeatActorActor = this.getContext().spawn(WorkerHeartbeatActor.create(bxJobServerList), "bx-job-worker-WorkerHeartbeatActor");
            workerHeartbeatActorActor.tell(WorkerHeartbeatActor.WorkerHeartbeatStartMsg.builder().build());
            return this;
        }).build();
    }

    private Behavior<WorkerRegistActorMsg> onStartRegistMsg(WorkerRegistStartMsg workerRegistStartMsg) throws InterruptedException {
        if (workerHeartbeatActorActor != null) {
            log.info("Currently registered, do not re-register");
            return this;
        }
        if (currentIndex == this.bxJobServerList.size()) {
            log.info("current no available server , please check network  server list :{}", this.bxJobServerList);
            currentIndex = 0;
            // 3s 后继续重试
            JobWorkerBootstrap.getGlobalScheduledExecutorService().schedule(() -> this.getContext().getSelf().tell(WORKER_REGIST_START_MSG), 3, TimeUnit.SECONDS);
            return this;
        }
        Gson gson = new Gson();
        String currentServer = this.bxJobServerList.get(currentIndex++);
        WorkerRegistRequest workerRegistRequest = new WorkerRegistRequest();
        //todo 组装 WorkerRegistRequest 信息
        HttpRequest httpRequest = HttpRequest.create(currentServer + REGIST_PATH).withMethod(HttpMethods.POST).withEntity(HttpEntities.create(ContentTypes.APPLICATION_JSON, gson.toJson(workerRegistRequest)));
        final CompletionStage<HttpResponse> responseFuture = Http.get(this.getContext().getSystem()).singleRequest(httpRequest);
        responseFuture.whenComplete((response, error) -> {
            if (error != null) {
                log.error("regist failed, server:{}", currentServer, error);
                this.getContext().getSelf().tell(WORKER_REGIST_START_MSG);
                return;
            }
            CompletionStage<CommonResponse> unmarshal = Jackson.unmarshaller(CommonResponse.class).unmarshal(response.entity(), this.getContext().getSystem());
            getContext().pipeToSelf(unmarshal.toCompletableFuture(), (res, throwable) -> {
                if (throwable == null && res.getStatus() == CommonResponse.SUCCESS_STATUS) {
                    log.info("regist success, server:{}", currentServer);
                    return WorkerRegistSuccessMsg.builder().build();
                } else {
                    log.info("regist failed, server:{},response:{}", currentServer, res, throwable);
                    return WORKER_REGIST_START_MSG;
                }

            });
        });
        return this;
    }
}